package cn.doitedu.service;

import cn.doitedu.dao.HistoryEventDao;
import cn.doitedu.dao.HistoryEventStateDaoImpl;
import cn.doitedu.dao.ProfileDao;
import cn.doitedu.pojo.CombineCondition;
import cn.doitedu.pojo.EventStateBean;
import cn.doitedu.pojo.LogBean;
import cn.doitedu.pojo.RuleCondition;
import cn.doitedu.utils.EventComparator;
import cn.doitedu.utils.MatchUtils;
import cn.doitedu.utils.TimeBoundUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.java.utils.ParameterTool;

import java.util.List;


/**
 * @create: 2021-11-12 19:21
 * @author: 静如海的男人
 * @program: eagle-text
 **/
@Slf4j
public class QueryService {
    //查询Hbase
    private ProfileDao profileDao;
    //查询ClickHouse
    private HistoryEventDao historyEventDao;
    //查询状态
    private HistoryEventStateDaoImpl historyEventStateDao;

    //初始化
    public void init(ParameterTool parameterTool, ListState<EventStateBean> eventListState) throws Exception {
        //通过反射创建实例  匹配用户画像
        String profileDaoClazz = parameterTool.getRequired("profile.dao.impl");
        profileDao = (ProfileDao) Class.forName(profileDaoClazz).newInstance();
        profileDao.init(parameterTool);

        //查询状态返回历史行为数据
        historyEventStateDao = new HistoryEventStateDaoImpl(eventListState);

        //通过反射创建实例  匹配用户历史行为
        String historyEventDaoClazz = parameterTool.getRequired("history.event.dao.impl");
        historyEventDao = (HistoryEventDao) Class.forName(historyEventDaoClazz).newInstance();
        historyEventDao.init(parameterTool);

    }

    public boolean isMatch(LogBean bean, RuleCondition rcn) throws Exception {

        //1.匹配当前行为数据 判断是否定义行为规则
        boolean isMatchTriggerEvent = EventComparator.isMatchTriggerEvent(bean, rcn.getTriggerEvent());
        if (!isMatchTriggerEvent) {
            return false;
        }


        //2.匹配用户画像{tag1 = v1, tag10 = v1}
        boolean isMatchProfile = profileDao.isMatchProfile(bean, rcn.getProfileCondition());
        if (!isMatchProfile) {
            return false;
        }


        //3.匹配历史行为数据（匹配行为次数）
        boolean isMatchHistoryEvent = this.isMatchHistoryEvent(bean, rcn.getCombineCondition());
        if (!isMatchHistoryEvent) {
            return false;
        }

        return true;
    }


    private boolean isMatchHistoryEvent(LogBean bean, List<CombineCondition> combineConditions) throws Exception {

        for (CombineCondition combineCondition : combineConditions) {
            //根据数据中携带的timeStamp计算查询时间边界
            long timeBound = TimeBoundUtil.getTimeBound(bean.getTimeStamp());
            //事先定义的开始时间
            long startTime = combineCondition.getStartTime();
            //事先定义的结束时间
            long endTime = combineCondition.getEndTime();

            if (timeBound <= startTime) {

                //如果边界时间小于开始时间那么就查状态
                String sequenceStr = historyEventStateDao.queryEventSequenceStr(bean, combineCondition, startTime, endTime);
                int i = MatchUtils.matchCount(sequenceStr, combineCondition.getRegex());
                if (i < combineCondition.getMinLimit()) return false;

            } else if (timeBound >= endTime) {

                //如果边界时间大于结束时间那么就查ClickHouse
                String sequenceStr = historyEventDao.queryEventSequenceStr(bean, combineCondition, startTime, endTime);
                int i = MatchUtils.matchCount(sequenceStr, combineCondition.getRegex());
                if (i < combineCondition.getMinLimit()) return false;

            } else {

                //跨时间边界 先到状态中查询  再到clickHouse中查询
                //查询状态
                String sequenceStr1 = historyEventStateDao.queryEventSequenceStr(bean, combineCondition, timeBound, endTime);
                int i1 = MatchUtils.matchCount(sequenceStr1, combineCondition.getRegex());
                if (i1 < combineCondition.getMinLimit()) {
                    //查询ClickHouse
                    String sequenceStr2 = historyEventDao.queryEventSequenceStr(bean, combineCondition, startTime, timeBound);
                    String sequenceStr = sequenceStr2 + sequenceStr1;
                    int i2 = MatchUtils.matchCount(sequenceStr, combineCondition.getRegex());
                    if (i2 < combineCondition.getMinLimit()) {
                        return false;
                    }
                }
            }
        }

        return true;
    }

    //释放资源
    public void close() throws Exception {
        profileDao.close();
        historyEventDao.close();

    }


}
